package com.sita.redsun.mq;

import com.google.common.collect.Lists;
import com.sita.redsun.connector.handler.EchoServerHandler;
import com.sita.redsun.connector.server.NodeMaping;
import org.apache.activemq.command.ActiveMQTextMessage;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;

import javax.annotation.PostConstruct;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;
import org.apache.activemq.spring.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by zhebinwang on 2017/3/29.
 */
@Component("mQReciveListener")
public class MQReciveListener  implements MessageListener {
    private static final Logger logger = LoggerFactory.getLogger(MQReciveListener.class.getName());
    private static ThreadPoolExecutor pool;



    @PostConstruct //构造之后自动执行这个方法
    public void init() {
        logger.info("init QueueReceiver");

            BlockingQueue<Runnable> queue = new LinkedTransferQueue<Runnable>() {
                @Override
                public boolean offer(Runnable e) {
                    return tryTransfer(e);
                }
            };

            int tpCore = Integer.valueOf(System.getProperty("threadpool.core.size", "8"));
            int tpMax = Integer.valueOf(System.getProperty("threadpool.max.size", "100"));
           // logger.info("DSPT_THREAD_POOL core-size={}, max-size={}", 2, 4);
            pool = new

                ThreadPoolExecutor(
                        tpCore, tpMax,
                    1L,TimeUnit.MILLISECONDS,
                        queue) ;


            pool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                    try {
                        executor.getQueue().put(r);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
            });
        }





    //mq的消息从这里接收过来，这是因为在类 MQRecive中进行了配置
    @Override
    public void onMessage(Message message) {
        System.out.println("接收到的消息为："+message.toString());
        //收到消息后调用rest接口发送
        ActiveMQTextMessage mqTextMessage=(ActiveMQTextMessage)message;
        String str=null;
        try {
             str=mqTextMessage.getText();
        } catch (JMSException e) {
            e.printStackTrace();
        }
        tryRest(str);
        //message.get()

    }




    private void tryRest(final String  message) {

            pool.execute(new Runnable() {
                @Override
                public void run() {
                    handleRestSpring(message);
                    //handleUsingJersey(message);

                }
            });
            //logger.info("当前线程池的大小" + pool.getPoolSize());

    }



    private void handleRestSpring(String  message) {
       // logger.info("Begin forward MoMessage={} time={}", message.hashCode(), System.currentTimeMillis());
       // logger.info("DSPT_POOL activeCnt={}", pool.getActiveCount());

        String identifier = //message.getIdentifier();
                message.substring(8,24);
        //String targetAddress = "http://10.105.222.201:8080/yadea/v1/messages/KKSMOD";
        int neednodeid=getNeedNodeId(identifier);
        String targetAddress = "http://"+NodeMaping.nodeMaping[neednodeid].getLanaddress()+":8080/yadea/v1/messages/KKSMOD";
        // Combine the url
        String requestUri = UriComponentsBuilder
                .fromUriString(targetAddress)
                .queryParam("imei", identifier)
                //.buildAndExpand(message.getProtocol())
                .toUriString();

       // logger.info("被分发的地址是:{}", requestUri);

        // Create the request header
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(org.springframework.http.MediaType.APPLICATION_JSON);
        headers.setAccept(Lists.newArrayList(org.springframework.http.MediaType.APPLICATION_JSON));

        headers.set("Connection", "Close");
        //headers.setAcceptCharset(Lists.newArrayList(StandardCharsets.UTF_8));
        ResponseEntity<String> response=null;
        try {
            HttpEntity<String> requestEntity = new HttpEntity<String>(message, headers);
            RestTemplate restTemplate = new RestTemplate();
            logger.info(requestUri);
            response = restTemplate.exchange(requestUri, HttpMethod.POST, requestEntity, String.class);

            if (response.getStatusCode().is2xxSuccessful()) {
                //logger.info("分发成功 {}", message);
            } else {
                //logger.info("分发失败 {}, 详情 {}", requestUri, response.getStatusCode().getReasonPhrase());
            }
        } catch (RestClientException e) {
            e.printStackTrace();
            logger.info("发送错误");
            if (response!=null)
            {
                logger.info(response.toString());
            }
        }
    }

    /**需要发送到哪个服务器节点***/
    private int getNeedNodeId(String identifier){

        int neednodeid= Long.valueOf(Long.valueOf(identifier) % NodeMaping.nodeMaping.length).intValue();

        return neednodeid;
    }
}
